#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netinet/tcp.h>

#include<iostream>

using std::cout;
using std::endl;

#include <log4cxx/xml/domconfigurator.h>
#include <event2/thread.h>
#include <event2/listener.h>

#include "ServerFrameWork.h"

LoggerPtr ServerFrameWork::_logger(Logger::getLogger("ServerFrameWork"));

int ServerFrameWork::init_logger(const char * log_cnf_file) {
	if (NULL == log_cnf_file) return -1;
	setlocale(LC_ALL, "zh_CN.GBK");
	_log_cnf_file = log_cnf_file;
	xml::DOMConfigurator::configure(_log_cnf_file.c_str());
	return 1;
}


int ServerFrameWork::init(int thread_number,  HANDLE_SOCKET_INPUT_FUNC func, void * arg, \
			const char * listen_ip, unsigned short listen_port, const int socket_idle_timeout, \
			THREAD_INIT_FUNC thread_init_func, void * thread_init_func_arg) {
	if (thread_number < 1 || NULL == func || NULL == listen_ip || listen_port < 1) {
		LOG4CXX_ERROR(_logger, "start server error, input args is not valid");
		return -2;
	}
	_thread_number = thread_number;
	_handle_socket_input_func = func;
	_handle_socket_input_func_arg = arg;
	_listen_ip = listen_ip;
	_listen_port = listen_port;
	_socket_idle_timeout = socket_idle_timeout;
	_thread_init_func = thread_init_func;
	_thread_init_func_arg = thread_init_func_arg;


	evthread_use_pthreads();
	_eb = event_base_new();
	if (NULL == _eb) {
		LOG4CXX_ERROR(_logger, "could not create libevent's event base");
		return -3;
	}
	pthread_mutex_init(&_active_sockets_mutex, NULL);
	pthread_cond_init(&_worker_thread_wake_up_cond, NULL);
	LOG4CXX_INFO(_logger, "init success!");
}

int ServerFrameWork::run_server_loop() {
	int ret = event_base_loop(_eb, 0);
	LOG4CXX_ERROR(_logger, "evnet loop ended with ret:" << ret);
	return ret;
}

int ServerFrameWork::start_server() {
	int ret = set_up_listner();
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, "start server error, could not set up listner, ret:" << ret);
		return -2;
	}
	pthread_t * thread = new pthread_t[_thread_number];
	for (int i = 0; i < _thread_number; i++) {
		ret = pthread_create(&thread[i], NULL, worker_thread_routine, this);
		if (0 != ret) {
			delete[] thread;
			LOG4CXX_ERROR(_logger, "start server error, cound not create thread, ret:" << ret << ", i:" << i);
			return -3;
		}
	}
	ret = run_server_loop();
	if (ret < 0) {
		delete[] thread;
		LOG4CXX_ERROR(_logger, "start server error, server loop ret:" << ret);
		return -4;
	}
	for (int i = 0; i < _thread_number; i++) {
		pthread_join(thread[i], NULL);
		LOG4CXX_INFO(_logger, "thread terminated, thread id:" << thread[i]);
	}
	LOG4CXX_INFO(_logger, "whole server terminated");
	return 0;
}

int ServerFrameWork::set_up_listner() {
	struct sockaddr_in sin;
	memset(&sin, 0, sizeof(sin));
	/* This is an INET address */
	sin.sin_family = AF_INET;
	struct in_addr inaddr;
	int ret = inet_aton(_listen_ip.c_str(), &inaddr);
	if (0 == ret) {
		LOG4CXX_ERROR(_logger, "set up listner error, invalid listen ip address:" << _listen_ip);
		return -1;
	}
	sin.sin_addr = inaddr;
	/* Listen on the given port. */
	sin.sin_port = htons(_listen_port);

	_listner = evconnlistener_new_bind(_eb, listner_cb, this, LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE, -1, \
			(struct sockaddr*)&sin, sizeof(sin));
	if (NULL == _listner) {
		LOG4CXX_ERROR(_logger, "create listner error");
		return -2;
	}
	LOG4CXX_INFO(_logger, "set up lisnter ok, now listen at ip:" << _listen_ip << ",  port:" << _listen_port);
	return 0;	
}

void ServerFrameWork::listner_cb(struct evconnlistener *listener, evutil_socket_t fd, struct sockaddr *address, \
                        int socklen,  void *ctx) {
	ServerFrameWork * sfw = (ServerFrameWork *)ctx;
	struct sockaddr_in * sin = (struct sockaddr_in *)address;
	char ip[1024];
	bzero(ip, sizeof(ip));
	int len;
	inet_ntop(AF_INET, &sin->sin_addr, ip, len);
	int port = ntohs(sin->sin_port);
	LOG4CXX_DEBUG(sfw->_logger, "accepted socket from ip:" << ip << ", port:" << port << ", fd:" << fd << \
		", predefined idle timeout:" << sfw->_socket_idle_timeout);
	int no_delay_flag = 1;
        setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &no_delay_flag,  sizeof(no_delay_flag));
	sfw->add_socket_to_event_base(fd);
}

void  ServerFrameWork::socket_event_cb(evutil_socket_t fd, short events, void * arg) {
	ServerFrameWork * sfw = (ServerFrameWork *)arg;
	if (EV_TIMEOUT == (events & EV_TIMEOUT)) {
		sfw->handle_socket_idle_timeout(fd);
	}
	else if (EV_READ == (events & EV_READ)) {
		LOG4CXX_DEBUG(sfw->_logger, "incoming reading event on fd:" << fd);
		sfw->add_active_socket(fd);
	}
	else {
		LOG4CXX_ERROR(sfw->_logger, "unknow events:" << events << " on socket fd:" << fd);
	}
}

int ServerFrameWork::add_active_socket(int fd) {
	if (fd < 1) {
		LOG4CXX_ERROR(_logger, "add active socket error, fd is not valid");
		return -1;
	}
	pthread_mutex_lock(&_active_sockets_mutex);
	_active_sockets.push_back(fd);
	pthread_cond_signal(&_worker_thread_wake_up_cond);
	pthread_mutex_unlock(&_active_sockets_mutex);
	return 1;
}

int ServerFrameWork::pop_active_socket() {
	int fd(-1);
	pthread_mutex_lock(&_active_sockets_mutex);
	while (_active_sockets.size() < 1) {
		pthread_cond_wait(&_worker_thread_wake_up_cond, &_active_sockets_mutex);
	}
	list<int>::const_iterator iter = _active_sockets.begin();
	fd = *iter;
	_active_sockets.pop_front();
	pthread_mutex_unlock(&_active_sockets_mutex);
	return fd;
}

int ServerFrameWork::handle_socket_idle_timeout(int fd) {
	if (fd < 1) {
		LOG4CXX_ERROR(_logger, "handle socket idle timeout error, invalid socket, fd:" << fd);
		return -1;
	}
	LOG4CXX_DEBUG(_logger, "socket idle timeout, fd:" << fd);
	close(fd);
	return 0;
}

void * ServerFrameWork::worker_thread_routine(void * arg) {
	ServerFrameWork * sfw = (ServerFrameWork *)arg;
	if (NULL != sfw->_thread_init_func) (*sfw->_thread_init_func)(sfw->_thread_init_func_arg);
	while (true) {
		int fd = sfw->pop_active_socket();
		int ret = (*(sfw->_handle_socket_input_func))(fd, sfw->_handle_socket_input_func_arg);
		if (ret < 1) {
			LOG4CXX_ERROR(sfw->_logger, "close socket, fd:" << fd << ", because handle socket input func ret:" << ret);
			close(fd);
		}
		/*
		int serr(0);
		socklen_t len = sizeof(serr);
                getsockopt(fd, SOL_SOCKET, SO_ERROR, &serr, &len);
		if (0 != serr) {
			LOG4CXX_ERROR(sfw->_logger, "found socket error, after handle socket input event, fd:" << fd << \
					", err code:" << serr);
			close(fd);
		}
		*/
		else sfw->add_socket_to_event_base(fd);
	}
}

int ServerFrameWork::add_socket_to_event_base(int fd) {
	if (fd < 1) {
		LOG4CXX_ERROR(_logger, "add socket to event base error, invalid socket fd:" << fd);
		return -1;
	}
	short events = EV_READ;
	struct timeval  * ptv = NULL;
	struct timeval tv;
	if (_socket_idle_timeout > 0) {
		events |= EV_TIMEOUT;
		tv.tv_sec = _socket_idle_timeout;
		tv.tv_usec = 0;
		ptv = &tv;
	}
	int ret = event_base_once(_eb, fd, events, socket_event_cb, this, ptv);
	if (0 != ret) {
                LOG4CXX_ERROR(_logger, "add socket to event base error, fd:" << fd << ", predefined idle timeout:" << \
				_socket_idle_timeout);
		return -2;
        }
	return 0;
}
